package cn.baker.delayqueue.mapper;


import cn.baker.delayqueue.config.DelayQueueProperties;
import cn.baker.delayqueue.entity.DelayQueueMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.List;

/**
 * @author yuanyu
 */
@Slf4j
public class DelayQueueMessageMapper {

    private final JdbcTemplate jdbcTemplate;
    private final DelayQueueProperties delayQueueProperties;

    public DelayQueueMessageMapper(JdbcTemplate jdbcTemplate, DelayQueueProperties delayQueueProperties) {
        this.jdbcTemplate = jdbcTemplate;
        this.delayQueueProperties = delayQueueProperties;
    }

    /**
     * 插入消息
     *
     * @param message 消息
     */
    public int insert(DelayQueueMessage message) {
        Object[] args = { message.getQueueCode(), message.getDelayInMillis(), message.getPayload() };
        String sql = "insert into delay_queue_message (id, queue_code, delay_in_millis, payload)\n" +
                "values (null, ?, ?, ?);";
        return jdbcTemplate.update(sql, args);
    }

    /**
     * 扫描消息
     *
     * @return
     */
    public List<DelayQueueMessage> scanMessage(long delayInMillis, int from, int limit) {
        Object[] args = {delayInMillis, from, limit};
        String sql = "select `id`, `queue_code` as queueCode, `delay_in_millis` as delayInMillis, `payload`\n" +
                     "from delay_queue_message\n" +
                     "where delay_in_millis <= ?\n" +
                     "order by delay_in_millis asc\n" +
                     "limit ?, ? for update;";

        List<DelayQueueMessage> result =  jdbcTemplate.query(sql, new BeanPropertyRowMapper<>(DelayQueueMessage.class), args);
        return result;
    }


    public void deleteByIds(List<Long> pidList) {
        for (Long pid : pidList) {
            Object[] args = { pid };
            String sql = "delete from delay_queue_message where id = ?;";
            int count = jdbcTemplate.update(sql, args);
            log.info("deleteById id={},count={}", pid, count);
        }
    }
}
